CasaTrade 배치와 API가 도메인 로직을 공유하게 만들기

목차

• #같은 가격 규칙이 두 군데에 생기는 순간
• #무엇을 공유하고 무엇을 분리할 것인가
• #순수한 도메인 함수로 가격 결정을 표현하기
• #애플리케이션 유스케이스가 의존성을 조립하기
• #API 어댑터의 책임
• #배치 실행기의 책임
• #한 건 유스케이스와 대량 처리의 관계
• #트랜잭션 경계를 호출자가 선택하게 하기
• #재실행 가능한 배치와 체크포인트
• #모듈 import 시 작업을 시작하지 않기
• #재구성한 Python 예시
• #규칙 버전과 결과 재현성
• #테스트 전략과 운영 지표
• #결론
• #관련 노트

같은 가격 규칙이 두 군데에 생기는 순간

관리자가 상품 하나의 가격을 다시 계산하는 HTTP API와 매일 전체 상품을 갱신하는 배치가 있다고 하자.

# API
price = median(rows)
if len(rows) < 3:
    confidence = 0.2

# batch
filtered = remove_outliers(rows)
price = percentile(filtered, 0.5)
if len(filtered) < 5:
    confidence = 0.2

처음에는 비슷하지만 시간이 지나면 한쪽에만 조건이 추가된다.

• 통화 환산 기준일이 다르다.
• 판매 완료와 판매 중 가격을 다르게 포함한다.
• 이상치 제거 기준이 다르다.
• 증거가 적을 때 한쪽은 0, 다른 쪽은 null을 저장한다.
• 수동 실행과 야간 배치가 서로 다른 상태값을 만든다.

사용자는 같은 상품인데 “새로고침” 직후 가격과 다음 날 배치 가격이 달라지는 현상을 본다. 두 코드의 결과를 맞추기 위해 복사·붙여넣기를 반복하면 차이는 더 커진다.

무엇을 공유하고 무엇을 분리할 것인가

flowchart LR
    HTTP[HTTP API Adapter] --> U[Refresh Price Use Case]
    CLI[Batch CLI Adapter] --> U
    JOB[Queue Worker] --> U
    U --> D[Price Domain]
    U --> R[Repositories]
    U --> C[Clock and Policy]

공유할 것:

• 시장 가격 입력의 도메인 타입
• 통화와 상태 정규화 규칙
• 이상치·증거 선택 규칙
• 중앙값, 가격 범위, 신뢰도 계산
• 결과 상태와 실패 사유
• 결과 저장과 이벤트 발행 유스케이스

분리할 것:

• HTTP 요청 파싱과 응답 코드
• 배치 대상 조회와 chunk 크기
• 사용자 인증과 관리자 권한
• cron 스케줄
• CLI 출력과 progress 표시
• 재시도·속도 제한·체크포인트

API DTO나 웹 프레임워크의 Request 객체를 도메인 함수에 넘기면 배치가 HTTP 계층에 의존한다. 반대로 배치의 전역 DB connection을 도메인 안에서 import하면 단위 테스트와 API 트랜잭션 조립이 어렵다.

순수한 도메인 함수로 가격 결정을 표현하기

도메인 입력을 먼저 만든다.

from dataclasses import dataclass
from datetime import datetime
from decimal import Decimal
from typing import Literal

@dataclass(frozen=True)
class MarketEvidence:
    source_id: str
    amount_krw: Decimal
    status: Literal["available", "sold", "unknown"]
    observed_at: datetime
    similarity: float

@dataclass(frozen=True)
class PricePolicy:
    minimum_evidence: int
    minimum_similarity: float
    lower_quantile: float
    upper_quantile: float
    version: str

결과도 단순 숫자가 아니라 판단 근거를 포함한다.

@dataclass(frozen=True)
class PriceDecision:
    status: Literal["estimated", "insufficient"]
    center_price_krw: Decimal | None
    lower_price_krw: Decimal | None
    upper_price_krw: Decimal | None
    evidence_ids: tuple[str, ...]
    confidence: float
    policy_version: str
    reason: str

순수 함수는 DB, 현재 시각과 환경 변수를 직접 읽지 않는다.

def estimate_price(
    evidence: list[MarketEvidence],
    policy: PricePolicy,
    as_of: datetime,
) -> PriceDecision:
    usable = [
        item for item in evidence
        if item.similarity >= policy.minimum_similarity
        and item.observed_at <= as_of
        and item.amount_krw > 0
    ]

    if len(usable) < policy.minimum_evidence:
        return PriceDecision(
            status="insufficient",
            center_price_krw=None,
            lower_price_krw=None,
            upper_price_krw=None,
            evidence_ids=tuple(item.source_id for item in usable),
            confidence=0.0,
            policy_version=policy.version,
            reason="not_enough_evidence",
        )

    prices = sorted(item.amount_krw for item in usable)
    return build_quantile_decision(prices, usable, policy)

같은 evidence, policy와 as-of 시각이면 API와 배치 어디서 호출해도 같은 결과가 나온다.

애플리케이션 유스케이스가 의존성을 조립하기

순수 함수만으로는 데이터를 읽고 저장할 수 없다. 유스케이스가 repository와 트랜잭션을 조립한다.

class RefreshProductPrice:
    def __init__(
        self,
        products: ProductRepository,
        markets: MarketEvidenceRepository,
        prices: PriceEstimateRepository,
        policies: PricePolicyProvider,
        clock: Clock,
    ) -> None:
        self.products = products
        self.markets = markets
        self.prices = prices
        self.policies = policies
        self.clock = clock

    def execute(self, product_id: int) -> PriceDecision:
        product = self.products.require(product_id)
        evidence = self.markets.find_for(product)
        policy = self.policies.current_for(product.category)
        as_of = self.clock.now()

        decision = estimate_price(evidence, policy, as_of)
        self.prices.save(product.id, decision, as_of)
        return decision

이 클래스는 HTTP status나 argparse를 모른다. 저장소 구현도 protocol이나 추상 인터페이스 뒤에 둔다.

class MarketEvidenceRepository(Protocol):
    def find_for(self, product: Product) -> list[MarketEvidence]:
        ...

API 어댑터의 책임

API는 요청자 권한, 입력 형식, timeout과 응답 표현을 담당한다.

@router.post("/products/{product_id}/price:refresh")
def refresh_product_price(
    product_id: int,
    actor: AdminUser = Depends(require_price_admin),
    use_case: RefreshProductPrice = Depends(build_use_case),
):
    try:
        decision = use_case.execute(product_id)
        return present_price_decision(decision)
    except ProductNotFound:
        raise HTTPException(status_code=404)
    except RefreshAlreadyRunning:
        raise HTTPException(status_code=409)

관리자 버튼은 낮은 지연을 기대하므로 한 상품만 처리하고 진행 상태를 바로 반환할 수 있다. 계산이 오래 걸린다면 API가 job을 생성하고 202를 반환한 뒤 worker가 같은 유스케이스를 호출한다.

API가 estimate_price를 직접 호출한 뒤 각 repository를 따로 저장하면 애플리케이션 규칙이 다시 핸들러에 새어 나온다.

배치 실행기의 책임

배치는 대상을 선택하고 chunk를 나누며 rate limit과 실패 격리를 담당한다.

def run_price_refresh_batch(
    candidates: CandidateRepository,
    use_case_factory: Callable[[], RefreshProductPrice],
    checkpoint: BatchCheckpoint,
    batch_size: int,
) -> BatchStats:
    stats = BatchStats()

    while True:
        rows = candidates.claim_after(
            checkpoint.last_product_id,
            limit=batch_size,
        )
        if not rows:
            break

        for row in rows:
            try:
                use_case_factory().execute(row.product_id)
                stats.succeeded += 1
            except DomainError as error:
                stats.failed += 1
                record_item_failure(row.product_id, error)

            checkpoint.advance(row.product_id)

    return stats

배치가 도메인 예외를 무조건 삼키면 안 된다. 데이터 부족처럼 정상적인 도메인 결과와 DB 연결 실패 같은 인프라 오류를 구분한다. 인프라 전체가 장애면 항목마다 실패 로그를 수백만 개 만들기보다 circuit breaker로 배치를 멈춘다.

한 건 유스케이스와 대량 처리의 관계

한 건용 execute()를 100만 번 호출하면 매번 policy와 connection을 읽어 비효율적일 수 있다. 그렇다고 배치만을 위한 별도 가격 규칙을 만들지는 않는다.

class RefreshPriceChunk:
    def execute(self, product_ids: list[int]) -> list[RefreshResult]:
        policy_snapshot = self.policies.snapshot()
        products = self.products.find_many(product_ids)
        evidence_by_product = self.markets.find_many(products)

        results = []
        for product in products:
            decision = estimate_price(
                evidence_by_product.get(product.id, []),
                policy_snapshot.for_category(product.category),
                self.clock.now(),
            )
            results.append(RefreshResult(product.id, decision))

        self.prices.save_many(results)
        return results

공유해야 하는 핵심 계산 함수는 유지하고 I/O만 bulk화한다. 한 건과 chunk 경로가 같은 fixture에서 같은 PriceDecision을 내는 contract test를 둔다.

트랜잭션 경계를 호출자가 선택하게 하기

repository가 메서드마다 자동 commit하면 여러 저장을 하나로 묶기 어렵다. Unit of Work를 전달할 수 있다.

class RefreshProductPrice:
    def execute(
        self,
        product_id: int,
        unit_of_work: UnitOfWork,
    ) -> PriceDecision:
        with unit_of_work:
            product = unit_of_work.products.require(product_id)
            evidence = unit_of_work.markets.find_for(product)
            decision = estimate_price(
                evidence,
                self.policies.current_for(product.category),
                self.clock.now(),
            )
            unit_of_work.prices.save(product.id, decision)
            unit_of_work.outbox.add(
                PriceEstimateUpdated.from_decision(product.id, decision)
            )
            unit_of_work.commit()
            return decision

API는 한 상품 트랜잭션을, 배치는 항목별 또는 작은 chunk별 트랜잭션을 선택한다. 전체 배치를 하나의 트랜잭션으로 묶으면 잠금과 undo log가 커지고 한 건 실패로 모든 작업이 롤백된다.

외부 검색이나 이미지 모델 호출은 DB 트랜잭션 안에서 오래 기다리지 않는다. 증거 수집 단계와 결정 저장 단계를 분리하고 입력 snapshot ID를 남긴다.

재실행 가능한 배치와 체크포인트

배치는 중간에 죽고 다시 실행될 수 있다. OFFSET 기반 페이지는 처리 중 데이터가 바뀌면 누락과 중복이 생길 수 있다. 안정적인 키 커서를 사용한다.

SELECT product_id
FROM price_refresh_candidates
WHERE product_id > :last_product_id
ORDER BY product_id
LIMIT :batch_size;

가격 저장은 (product_id, as_of_date, policy_version) 같은 자연스러운 유일 키로 upsert할 수 있다. 재실행이 같은 결과를 중복 생성하지 않게 한다.

@dataclass
class BatchRun:
    run_id: str
    policy_version: str
    input_snapshot_id: str
    last_product_id: int | None
    status: Literal["running", "completed", "failed"]

체크포인트를 항목 처리 전에 앞으로 옮기면 실패 항목을 건너뛴다. 성공적으로 커밋한 뒤 이동한다. 같은 항목을 한 번 더 처리할 가능성은 허용하고 저장을 멱등하게 만드는 편이 안전하다.

모듈 import 시 작업을 시작하지 않기

배치 파일을 import하는 순간 DB 연결과 실행이 시작되면 API 테스트나 공유 함수 import가 부작용을 만든다.

# 좋지 않은 예시
connection = connect_database()
run_all_products(connection)

entry point를 분리한다.

def main() -> int:
    settings = load_settings()
    container = build_container(settings)
    stats = run_price_refresh_batch(
        candidates=container.candidates,
        use_case_factory=container.refresh_price,
        checkpoint=container.checkpoint,
        batch_size=settings.batch_size,
    )
    return 0 if stats.failed == 0 else 1

if __name__ == "__main__":
    raise SystemExit(main())

모듈 import는 타입과 함수 정의만 로드한다. cron, CLI와 worker가 명시적으로 main이나 handler를 호출한다.

재구성한 Python 예시

폴더 경계는 다음처럼 둘 수 있다.

price_service/
  domain/
    evidence.py
    estimate.py
    policy.py
  application/
    refresh_price.py
    refresh_chunk.py
  ports/
    repositories.py
    clock.py
  adapters/
    mariadb.py
    elasticsearch.py
    http.py
  entrypoints/
    api.py
    batch.py

도메인 패키지는 adapters를 import하지 않는다. 의존 방향을 테스트할 수도 있다.

def test_domain_does_not_import_infrastructure():
    forbidden = {"sqlalchemy", "requests", "fastapi"}
    imports = collect_imports("price_service/domain")
    assert imports.isdisjoint(forbidden)

동일 입력에 대한 contract test:

def test_api_and_batch_use_same_price_decision(fixture):
    api_result = fixture.api.refresh(fixture.product_id)
    batch_result = fixture.batch.refresh_one(fixture.product_id)

    assert api_result.center_price_krw == batch_result.center_price_krw
    assert api_result.evidence_ids == batch_result.evidence_ids
    assert api_result.policy_version == batch_result.policy_version

실제로는 두 실행 사이 데이터가 바뀌지 않도록 같은 snapshot repository와 clock을 사용한다.

규칙 버전과 결과 재현성

코드를 공유해도 오늘 실행과 한 달 전 실행은 입력과 정책이 다를 수 있다. 결과에 재현 정보를 남긴다.

{
  "productId": 42,
  "status": "estimated",
  "centerPriceKrw": 128000,
  "rangeKrw": [109000, 145000],
  "evidenceSnapshotId": "snapshot-example-7",
  "policyVersion": "price-policy-12",
  "calculatorVersion": "estimate-5",
  "calculatedAt": "2026-07-31T03:00:00Z"
}

모델이나 임계값을 바꿀 때 이전 결과와 새 결과를 shadow 계산해 비교할 수 있다. API만 새 버전을 쓰고 배치가 옛 버전을 쓰지 않도록 배포 artifact와 설정 버전을 관측한다.

환경 변수로 임계값을 즉시 바꾼다면 어느 실행이 어느 값을 사용했는지 snapshot을 남긴다. 그렇지 않으면 결과가 달라져도 원인을 설명할 수 없다.

테스트 전략과 운영 지표

테스트 계층:

• 순수 도메인 테스트: 이상치, 경계값, 증거 부족
• repository contract test: API용과 batch용 구현의 의미 일치
• 유스케이스 테스트: 저장과 outbox, 예외 처리
• API 테스트: 권한, DTO, status code
• 배치 테스트: chunk, checkpoint, 재실행
• golden fixture: 동일 snapshot의 버전별 결과 비교

결론

API와 배치가 가격 규칙을 각각 구현하면 작은 수정이 누적되어 같은 상품에 다른 결정을 내린다. 해결책은 HTTP 핸들러를 배치에서 호출하거나 배치 스크립트를 API에서 재사용하는 것이 아니다.

핵심은 입력 evidence와 policy로 PriceDecision을 만드는 순수 도메인 함수를 공유하고, repository와 트랜잭션을 조립하는 유스케이스를 둔 뒤, API와 배치는 인증·대상 선택·chunk·재시도와 출력만 담당하게 하는 것이다.

대량 처리는 I/O를 bulk화하되 같은 계산 규칙을 사용하고, 체크포인트와 유일 키로 재실행 가능하게 만든다. 입력 snapshot과 정책·계산기 버전을 결과에 남기면 두 실행 경로의 차이를 감으로 추적하지 않고 재현할 수 있다.

관련 노트

• Express 서버의 백그라운드 작업을 명시적으로 시작하기
• Outbox Pattern으로 DB와 이벤트 발행 일치시키기
• DTO 검증과 도메인 검증은 무엇이 다른가
• 서비스 계층의 트랜잭션 경계 정하기
• 장애 대응 Runbook에 들어가야 할 내용
• 베베스냅 육아 기록 일괄 입력 트랜잭션
• CasaTrade Elasticsearch 장애 시 검색 fallback 설계